package com.fdb.efp.nls.service.impl.rocketmq.handle;

import static com.fdb.efp.basic.framework.mq.constant.TopicEnums.QUEUE_TASK_HANG_TOPIC;

import java.io.UnsupportedEncodingException;
import java.util.concurrent.TimeUnit;

import com.fdb.efp.nls.service.vo.NlsApplyInfoVO;
import org.apache.rocketmq.common.message.MessageExt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.data.redis.core.StringRedisTemplate;

import com.alibaba.fastjson.JSONObject;
import com.fdb.basic.framework.core.threadpool.ApplicationThreadPool2;
import com.fdb.basic.framework.core.util.SpringContextUtils;
import com.fdb.efp.basic.framework.mq.handle.MQMessageProcessor;
import com.fdb.efp.nls.service.impl.queue.QueueHangHandle;

/**
 * @description:业务处理实现类
 * @date:2019年2月12日 下午2:26:50
 * @author:wangyk3
 */
public class QueueTaskHangMessageProcessorImpl implements MQMessageProcessor {
	private static Logger logger = LoggerFactory.getLogger(QueueTaskHangMessageProcessorImpl.class);
	private static final String prefix = "rc:" + QUEUE_TASK_HANG_TOPIC.getTopicId() + ":";
	
	public boolean handleMessage(MessageExt messageExt) {
		try {
			if (isRepeat(messageExt)) {
				logger.warn("消息重复发送！messageid: " + messageExt.getMsgId() 
					+ " key: " + messageExt.getKeys());
			} else {
				String topic = messageExt.getTopic();
				String tag = messageExt.getTags();
				String key = messageExt.getKeys();
				String body = new String(messageExt.getBody(), "UTF-8");
				logger.info("获取消息成功，消息主题Topic：【" + topic + "】,消息标签tag：【" + tag + "】,消息Key:【" + key + "】,消息体：【" + body + "】");
				NlsApplyInfoVO nlsApplyInfoVO = JSONObject.parseObject(body, NlsApplyInfoVO.class);
				//
				QueueHangHandle executoryQueueThread = (QueueHangHandle) SpringContextUtils.getBean("queueHangHandleService");
				executoryQueueThread.setNlsApplyInfoVO(nlsApplyInfoVO);
				try {
					threadpool.add(executoryQueueThread);
				} catch (Exception e) {
					logger.error(e.getMessage());
				}
			}
		} catch (UnsupportedEncodingException e) {
			logger.error(e.getMessage());
		} catch (Exception e) {
			logger.error(e.getMessage());
		}
		return true;
	}
	
	private boolean isRepeat(MessageExt messageExt) {
		String key = messageExt.getKeys();
		String checkKey = prefix + key;
		if (template.opsForValue().setIfAbsent(checkKey, "")) {
			template.expire(checkKey, 12, TimeUnit.HOURS);
			return false;
		}
		return true;
	}

	@Autowired
	@Qualifier("applicationThreadPool")
	private ApplicationThreadPool2 threadpool;

	private StringRedisTemplate template;

	public void setTemplate(StringRedisTemplate template) {
		this.template = template;
	}
}
